from concurrent.futures import ThreadPoolExecutor
from uuid import uuid1

from sqlalchemy import create_engine
import pandas as pd
from aibox.data.cache_tool import CacheKit
from sqlalchemy import types
import pymysql


def mem_usage(pandas_obj):
    """
    计算 DataFrame 内存占用量
    :param pandas_obj:
    :return:
    """
    if isinstance(pandas_obj, pd.DataFrame):
        usage_b = pandas_obj.memory_usage(deep=True).sum()
    else:  # we assume if not a df it's a series
        usage_b = pandas_obj.memory_usage(deep=True)
    usage_mb = usage_b / 1024 ** 2  # convert bytes to megabytes
    return "{:03.2f} MB".format(usage_mb)


def df_optimize(data):
    """
    对df中object数据进行处理，减少内存占用
    :param data:
    :return:
    """
    # print(mem_usage(data))
    gl_obj = data.select_dtypes(include=['object'])
    converted_obj = pd.DataFrame()
    for col in gl_obj.columns:
        num_unique_values = len(gl_obj[col].unique())
        num_total_values = len(gl_obj[col])
        if num_unique_values / num_total_values < 0.5:
            converted_obj.loc[:, col] = gl_obj[col].astype('category')
        else:
            converted_obj.loc[:, col] = gl_obj[col]
    data[converted_obj.columns] = converted_obj
    return data


def upper(df):
    return df.rename(columns=lambda x: x.upper())


def lower(df):
    return df.rename(columns=lambda x: x.lower())


# DataFrame扩展方法，保存to_sql时处理object类型数据，指定为VARCHAR
def pd_save(self, tablename, con, if_exists='append', index=False):
    dtyp = {c: types.VARCHAR(10)
            for c in self.columns[(self.dtypes == 'object') | (self.dtypes == 'category')].tolist()}
    return self.to_sql(tablename, con, if_exists=if_exists, index=index, dtype=dtyp)


pd.DataFrame.save = pd_save


# 通用线程池对象实例化
# executor = ThreadPoolExecutor(max_workers=10)
# 数据库操作线程池
# save_executor = ThreadPoolExecutor(max_workers=10)


# def thread_save(self, tablename, con, if_exists='append', index=False):
#     save_executor.submit(pd_save, self, tablename, con, if_exists=if_exists, index=index)


def cache_set(key, value):
    if isinstance(value, pd.DataFrame):
        CacheKit.cache.set(key, value)
    else:
        CacheKit.cache.cache.set(key, value)


def cache_get(key):
    result = CacheKit.cache.cache.get(key)
    if result is not None and not isinstance(result, pd.DataFrame) and not isinstance(result, str):
        try:
            result = result.decode()
        except Exception:
            try:
                result = pd.read_msgpack(result)
            except Exception:
                pass
    return result


def cache_del(key):
    CacheKit.cache.cache.delete(key)


def get_ms_time(series,format=None):
    """
    上海时区时间字符串
    :param series:
    :param format:
    :return:
    """
    dt = pd.to_datetime(series, errors='coerce', format=format)
    dt = dt.tz_localize('Asia/Shanghai')
    dt = int(dt.value / 10 ** 6)
    return dt


def to_pdtime_ms(series, tz=True):
    """
    :param series:pandas.Series
    :return:
    """
    series = pd.to_datetime(series, errors='coerce', unit='ms')
    if isinstance(series, pd.Series):
        series = series.dt.tz_localize('UTC')
        series = series.dt.tz_convert('Asia/Shanghai')
        if not tz:
            series = series.dt.tz_localize(None)
    else:
        series = series.tz_localize('UTC')
        series = series.tz_convert('Asia/Shanghai')
        if not tz:
            series = series.tz_localize(None)
    return series


def to_pdtime_str(series, format=None, tz=True):
    series = pd.to_datetime(series, errors='coerce', format=format)
    if tz:
        if isinstance(series, pd.Series):
            series = series.dt.tz_localize('Asia/Shanghai')
        else:
            series = series.tz_localize('Asia/Shanghai')
    return series


# 生成8位短uuid
uuidChars = ("a", "b", "c", "d", "e", "f",
             "g", "h", "i", "j", "k", "l", "m", "n", "o", "p", "q", "r", "s",
             "t", "u", "v", "w", "x", "y", "z", "0", "1", "2", "3", "4", "5",
             "6", "7", "8", "9", "A", "B", "C", "D", "E", "F", "G", "H", "I",
             "J", "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", "U", "V",
             "W", "X", "Y", "Z")


def short_uuid():
    uuid = str(uuid1()).replace('-', '')
    result = ''
    for i in range(0, 8):
        sub = uuid[i * 4: i * 4 + 4]
        x = int(sub, 16)
        result += uuidChars[x % 0x3E]
    return result


def get_short_uuids(length):
    uuid_list = [short_uuid() for i in range(length)]
    return uuid_list


def isEmpty(obj):
    """
    对象是否为空
    :param obj:
    :return:
    """
    return True if (obj is None or len(obj) == 0) else False


def get_cartesian_product(df_left, df_right):
    """
    生成两个df的笛卡尔积
    :return:
    """
    left = df_left.copy()
    right = df_right.copy()
    left['cartesian'] = 1
    right['cartesian'] = 1
    result = pd.merge(left, right, on='cartesian')
    result = result.drop('cartesian', axis=1)
    return result


# 暂定
def get_db(user, password, ip, instance='orcl', port=1521):
    engine_str = 'oracle://%s:%s@%s:%s/%s' % (user, password, ip, port, instance)
    engine = create_engine(engine_str, encoding='utf-8')
    return engine


# 暂定
def get_mysql(user, password, ip, db, port=3306):
    engine_str = f'mysql+pymysql://{user}:{password}@{ip}:{port}/{db}'
    engine = create_engine(engine_str, encoding='utf-8')
    return engine


# 暂定
def get_pg(user="postgres", password="ehl1234", ip="10.2.111.65", db="mydb", port=5432):
    engine_str = f'postgresql+psycopg2://{user}:{password}@{ip}:{port}/{db}'
    engine = create_engine(engine_str, encoding='utf-8')
    return engine
